package pt.inescid.components.jobmanager;

import java.io.File;
import java.io.IOException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;

import pt.inescid.components.api.Component;
import pt.inescid.components.api.JobManagerI;
import pt.inescid.components.commons.Resources;
import pt.inescid.components.messages.Task;
import pt.inescid.components.messages.TaskResult;
import pt.inescid.core.GridP2P;
import pt.inescid.utils.FileUtils;
import pt.inescid.utils.Log;
import rice.p2p.commonapi.Id;
import rice.p2p.commonapi.NodeHandle;
import rice.pastry.commonapi.PastryIdFactory;

/**
 * Job Manager. 
 * This component is responsible for creating and scheduling tasks in accordance with available
 * resources at the moment.
 * - The tasks, as divisions of input files, are distributed among available machines.
 * - After the computation of the tasks is completed, this module collects the results and builds the final output of the
 * respective application.
 * - In the inverse flow, the Job Manager is also responsible for receiving and computing
 * tasks from remote machines in accordance to the Application Adaptation Layer.
 * 
 * @author Filipe Paredes
 * @author Sergio Esteves (sesteves at gsd.inesc-id.pt)
 */
public class JobManager extends Component implements JobManagerI {
	
	//a = WholeInputSize=MinimumTaskSize
	/*
	 * The minimum task size is defined upon application registration, and it is the minimum size that would worth
	 * the task parallelization (i.e. assuming that all tasks have the same size)
	 */
	private long minTaskSize;
	//b = InputSize=MaximumTaskSize
	private long maxTaskSize;
	
	private String tempDir;

	private int jobHandlerCount = 0;
	private Object jobHdlCntLock = new Object();
	
	private int incomingTaskCount = 0;
	private Object incomTaskCntLock = new Object();
	
	private int tasksInProgress = 0;
	private Object tasksInProgressLock = new Object();
	
	// key: jobHandler value: JobHolder
	private Map<Integer,JobHolder> jobHolder = 
		new Hashtable<Integer,JobHolder>();
	
	// GiGi Architecture - 1st and 3rd layer	
	private PastryIdFactory localFactory;

	// Tasks submitted
	public Hashtable<Integer,Id> submitedTasks = new Hashtable<Integer,Id>();
	
	// Tasks that were already here
	private Hashtable<Integer,Integer> checkedTasks = new Hashtable<Integer,Integer>();	

	public int NUM_RETRIES = 200;
	public int AVG_HOPS = 10;
	public int AVG_PROXIMITY = 100;
	public int AVG_LOOKUP = 500;
	
	private int NMaxCachedTasks = Integer.MAX_VALUE;
	
	//Construction of the JobManager class.
	public JobManager(GridP2P gridP2P, long minTaskSize, long maxTaskSize, String tempDir) {
		super(gridP2P);
		this.minTaskSize = minTaskSize;
		this.maxTaskSize = maxTaskSize;					
		this.tempDir = tempDir;					
		log.write(this, "JobManager constructed", Log.FINER);
	}
	
	//Initialization of Current JobManager.
	public void init() {
		this.localFactory = new PastryIdFactory(gridP2P.getEnvironment());
	}
	
	private void createJobHolder(int jobHandler, String appID, int nTasks, 
			String configLocation, String outputLocation) {
		log.write(this, "createJobHolder called (jobHandler: " + jobHandler + 
				", appID: " + appID + ", nTasks: " + nTasks + ", outputLocation: " +
				outputLocation + ")", Log.FINER);
		if(!jobHolder.containsKey(jobHandler)) {
			JobHolder jh = new JobHolder(jobHandler, appID, nTasks, configLocation, outputLocation);
			jobHolder.put(jobHandler, jh);
		}
	}

	private JobHolder getJobHolder(int jobHandler) {
		log.write(this, "getJobHolder called (jobHandler: " + jobHandler + ")", Log.FINER);
		if(jobHolder.containsKey(jobHandler))
			return jobHolder.get(jobHandler); 
		return null;
	}
	
	private Resources calculateTaskCost(String appID, String configLocation, 
			String dataLocation, long dataSize, int nTasks) {
		log.write(this, "calculateTaskCost called", Log.FINER);
		
		Resources res = gridP2P.launchAppTrial(appID, configLocation, dataLocation);
		if(res == null) {
			log.write(this, "Error while launching application (trial)", Log.SEVERE);
			; //FIXME
			res = new Resources();
		}
		res.setCPU(0.1);
		res.setMem(0.1);
		res.setBW(0.1);
		res.setStorage(res.getStorage() + dataSize);
		
		return res;
	}	
	
	private byte[] getKey(byte[] config, byte[] data) {
		byte[] material; 
		if(data == null) {
			material = config;
		} else {
			material = new byte[config.length + data.length];
			System.arraycopy(config, 0, material, 0, config.length);
			System.arraycopy(data, 0, material, config.length, data.length);
		}
		try {
			MessageDigest algorithm = MessageDigest.getInstance("SHA-1");
			algorithm.reset();
			algorithm.update(material);
			return algorithm.digest();
		} catch (NoSuchAlgorithmException e) {
			log.write(this, "NoSuchAlgorithmException occurred: " + e.getMessage(), Log.SEVERE);
		}
		return null;
	}
	
	private boolean isTaskResultCached(byte[] config, byte[] data, int jobHandler, int n) {
		if(n > NMaxCachedTasks) return false;		
		final Id lookupKey = localFactory.buildId(getKey(config,data));
		byte[] result = gridP2P.lookupCache(lookupKey);
		if(result == null)
			return false;		
		gatherTaskResult(null, new TaskResult(jobHandler, n, result), -1);
		return true;
	}
	
	private List<Task> createTasks(int jobHandler, int nTasks, String appID, 
			String dataLocation, String configLocation, long dataSize) {
		log.write(this, "createTasks called (jobHandler: " +jobHandler + 
				", nTasks: " + nTasks + ", appID: " + appID + 
				", configLocation: " + configLocation + ", dataSize: " +
				dataSize + ")", 10);
		int n = nTasks;
		boolean calculatedTaskCost = false;
		Resources taskCost = null;
		List<Task> tasks = new ArrayList<Task>();
		byte[] data = null;		
		try {
			data = dataLocation == null? null : FileUtils.getDataFromFile(dataLocation);
		} catch (IOException e) {
			log.write(this, "IOException occurred: " + e.getMessage(), Log.SEVERE);
		} 
		NodeHandle source = gridP2P.getEndpoint().getLocalNodeHandle();
		
		while(n-- != 0) {
			byte[] config = gridP2P.buildConfig(appID, configLocation, n, nTasks);
			if(isTaskResultCached(config, data, jobHandler, n)) 
				continue;
			
			if(!calculatedTaskCost) {
				taskCost = 
					calculateTaskCost(appID, configLocation, dataLocation, dataSize, nTasks);				
				calculatedTaskCost = true;
			}
			Task t = new Task(source, jobHandler, n, taskCost, 
					appID, config, data, localFactory.buildId(config));
			tasks.add(t);
		}
		return tasks.isEmpty() ? null : tasks;
	}
	
	private List<String> toTMP(List<String> input) {
		List<String> newInput = new ArrayList<String>();
		for(String str : input) {
			if(!new File(str).exists()) {
				log.write(this, "input does not exist: " + str, Log.SEVERE);
				return null;
			}
			int lastIdx = str.lastIndexOf('/') + 1;
			String destination = tempDir + str.substring(lastIdx < 0 ? 0 : lastIdx);
			FileUtils.copyFile(str, destination);
			newInput.add(destination);
		}
		return newInput;
	}
	
	/**
	 * This method creates a job and handles the whole process 
	 * till the job is completed
	 */
	public int createJob(String appID, List<String> input, String output) {
		log.write(this, "createJob called (appID: " + appID + ", ouput: " + 
				output + ")", Log.FINER);		
		
		if((input = toTMP(input)) == null) {
			gridP2P.writeln("input source files not found!");
			return -1;
		}
		
		String dataLocation = gridP2P.getDataLocation(appID, input);
		String configLocation = gridP2P.getConfigLocation(appID, input);		
		if(dataLocation == null && configLocation == null) {
			gridP2P.writeln("The specified application is not supported!");
			return -1;
		}
		long dataSize = FileUtils.getFileSize(dataLocation == null? configLocation : dataLocation);

		if(dataSize < minTaskSize)
			//TODO the job is not created!
			;
		
		long limit = dataSize / minTaskSize; 			
		log.write(this, "Maximum number of tasks: " + limit, Log.FINER);
		limit = 100;//??????
		
		int nAvailableNodes = gridP2P.getNumAvailableNodes(appID, (int)limit) + 1;
		log.write(this, "Number of available nodes: " + nAvailableNodes, Log.FINE);
		
		int jobHandler;
		synchronized(jobHdlCntLock) {
			jobHandler = jobHandlerCount++;
		}
		createJobHolder(jobHandler, appID, nAvailableNodes, configLocation, output);
		gridP2P.createState(appID, jobHandler, configLocation);
		
		List<Task> tasks = 
			createTasks(jobHandler, nAvailableNodes, appID, dataLocation, configLocation, dataSize);	
		
		if(tasks == null) {
			log.write(this, "no tasks created", Log.FINE);
			return 0;
		}
		
		// a task for being processed by this node
		Task task = tasks.remove(0);
		if(!tasks.isEmpty())
			gridP2P.distributeTasks(appID, tasks, tasks.get(0).getCost());
		processTask(null, task, -1);		
		return 0;
	}
	
	/*
	public JobInfo createGridlets(double cpu, double mem, double bw, 
			int profile, String data) {

		Task g = new Task(cpu, mem, bw, data);
		synchronized (this.app.env) {
			g.setLookupKey(localFactory.buildId(g.getData()));
		}
		app.notifyGridletCreation(profile, String.valueOf(g.hashCode()));
		Id idDest = om.forwardMessage(null, g);
		setNodeSelected(g.hashCode(), idDest);


		File f = new File("stats\\" + g.hashCode() + ".txt");
		boolean gridletSent = false;
		long gridletSentTime = 0;


		// wait at least for the gridlet to be sent
		while (!gridletSent) {
			if (f.exists()) {
				gridletSentTime = app.env.getTimeSource().currentTimeMillis();
				gridletSent = true;
			} else {
				Thread.currentThread().yield();
			}
		}

		// (Note: proximity = rtt / 2)
		Double rtt = Double.valueOf(om.getProximity(idDest));
		long proximity = rtt.longValue() / 2;

		// TODO: analisar o cálculo da estimativa do tempo de processamento de uma gridlet
		long delay = proximity + ((long) ((10 - g.getCpu()) * g.getDataSize())) + AVG_LOOKUP;
		// END TODO

		long timeLimit = gridletSentTime + delay;
		long maxDelay = delay + ((AVG_PROXIMITY * AVG_HOPS) * NUM_RETRIES);

		return new JobInfo(this, g, timeLimit, g.getLookupKey(), maxDelay, NUM_RETRIES);
	}
*/

/*	
	public void testNetwork() {
		om.forwardToAll(null, 0);
	}

	public void upNumTest(int incr) {
		app.upNumTest(incr);
	}
*/

	public void updateReputation(Id id, int metric, int value) {
		//FIXME
		//om.updateReputation(id, metric, value);	
	}
	
	
	public int getNumTasksInProgress() {
		int number;
		synchronized(tasksInProgressLock) {
			number = tasksInProgress;
		}
		return number;
	}

	// testing purposes
	public void setNumTasksInProgress(int value) {
		synchronized(tasksInProgressLock) {
			this.tasksInProgress = value;
		}		
	}
	
	/**
	 * 
	 * Check if the node is available to process the task
	 * 
	 * @param id - from
	 * @param message
	 * @param currentTime
	 */
	public void processTask(Id id, Task message, long currentTime) {
		log.write(this, "processTask called (id: " + id + ", message: " +
				message + ", currentTime: " + currentTime + ")", 20);

		Id source = message.getSource();
		gridP2P.writeln("Processing task from node " + source + ".");
		
		synchronized(tasksInProgressLock) {	tasksInProgress++; }		
		try {
			byte[] config = message.getConfig();
			byte[] data = message.getData();

			String configLocation = null, dataLocation = null, appID = message.getAppID();
			synchronized(incomTaskCntLock) {
				if(config != null)
					configLocation = "GridP2P_config" + incomingTaskCount + gridP2P.getConfigExt(appID);				
				if(data != null)
					dataLocation = "GridP2P_data" + incomingTaskCount + gridP2P.getDataExt(appID);
				incomingTaskCount++;
			}
			if(!gridP2P.isStream(appID)) {
				FileUtils.generateFile(tempDir + dataLocation, data);
				FileUtils.generateTextFile(tempDir + configLocation, config);
			}
			// reduce resources
			gridP2P.updateResources(-1*message.getCPU(), -1*message.getMem(), 0);				
			// launch application
			String outputLocation;
			if(gridP2P.isStream(appID))
				outputLocation = gridP2P.launchApp(appID, config, configLocation);
			else
				outputLocation = gridP2P.launchApp(appID, configLocation, dataLocation); 
			if(outputLocation == null) {					
				log.write(this, "Error while lauching application - outputLocation is null"
						, Log.SEVERE);
				; //TODO error occurred
				synchronized(tasksInProgressLock) { tasksInProgress--; }
				gridP2P.writeln("Error while processing task!");
				return;
			}
			byte[] output = FileUtils.getDataFromFile(tempDir + outputLocation);						
			int jobHandler = message.getJobHandler();
			int taskNumber = message.getTaskNumber();
			if(taskNumber <= NMaxCachedTasks)
				gridP2P.cacheTaskResult(localFactory.buildId(getKey(config, data)), output);
			if(id == null) // own task 
				gatherTaskResult(id, new TaskResult(jobHandler, taskNumber, 
						output), currentTime);
			else
				gridP2P.sendResult(source, jobHandler, 
						taskNumber, output, message.getCPU());				
			gridP2P.updateResources();
		} catch (IOException e) {
			log.write(this, "IOException occurred: " + e.getMessage(), Log.SEVERE);
		}								

		synchronized(tasksInProgressLock) { tasksInProgress--; }
		gridP2P.writeln("Task from node " + source + " completed.");
		
		int n = 0;
		synchronized (checkedTasks) {
			// Verify if this task was already scanned
			if (checkedTasks.containsKey(message.hashCode())) {
				n = checkedTasks.get(message.hashCode());
				
				//FIXME
				//if (n == NUM_NEIGHBORS) {
					//checkedGridlets.remove(message.hashCode());
					//return;
				//}
			}

			if (checkedTasks.size() > 9999) {
				checkedTasks.remove(0);
			}

			checkedTasks.put(message.hashCode(), n + 1);
		}

		long processTime = 0;

		// must be synchronized to ensure that the params are updated when the 
		// availability is positive
		synchronized (this) {
			//processTime = checkAvailability(null, message);

			/* FIXME
			if (processTime != 0) {
				om.updateParams(getId(), message, true);
				Task t = (Task) message;
				app.addProcessToQueue(new JobInfo(this, t, currentTime+processTime,
						t.getLookupKey(), 0, 0));
			} else {
				gridP2P.forwardMessage(id, message);
			}
			*/
		}
	}

	public void gatherTaskResult(Id id, TaskResult message, long currentTime) {
		log.write(this, "gatherTaskResult called (id: " + id + ", message: " +
				message + ", currentTime: " + currentTime + ")", Log.INFO);											
		
		int jobHandler = message.getJobHandler();
		int nTask = message.getTaskNumber();
		JobHolder jh = getJobHolder(jobHandler);		
		if(jh.contains(nTask)) {
			log.write(this, "Task already stored!", Log.INFO);
			return;
		}		
		gridP2P.controlInbound(message);
		byte[] data = message.getData();
		jh.addTaskResult(nTask, data);
		String appID = jh.getAppID();
		gridP2P.updateState(appID, jobHandler, data);

		if(jh.isCompleted()) {
			String outputLocation = jh.getOutputLocation();
			byte[] output;
			if(!gridP2P.isStream(appID)) {
				output = gridP2P.buildOutput(appID, jh.getTaskResults());			
				try {	
					FileUtils.generateFile(outputLocation, output);
				} catch (IOException e) {
					log.write(this, "IOException occurred: " + e.getMessage(), Log.SEVERE);
				}
			} else {
				if(!gridP2P.buildOutput(appID, jobHandler, jh.getConfigLocation(), outputLocation)) {
					log.write(this, "Error while building output!", Log.SEVERE);
					gridP2P.writeln("Job error!");
					jobHolder.remove(jobHandler);
					return;
				}
			}
			log.write(this, "Job " + jobHandler + " is completed!", Log.INFO);
			jobHolder.remove(jobHandler);
			gridP2P.writeln("Job is complete! (output: " + outputLocation + ")");
		}
	}

	/* FIXME

	public void getInfo(int num, PrintStream stdout, PrintStream psTemp) {
		om.getInfo(num, stdout, psTemp);
	}

	public int getNodeType() {
		return om.getNodeType();
	}

	public Past getPastApp() {
		return pastApp;
	}
*/
	// Sets the 1st node selected for routing the gridlet with the 'hashcode'
	public void setNodeSelected(int hashCode, Id node) {
		submitedTasks.put(hashCode, node);
	}

	// Returns the 1st node selected for routing the gridlet with the 'hashcode'
	public Id getNodeSelected(int hashCode) {
		return submitedTasks.get(hashCode);
	}
/*
	private long getProcessingTime(Message message) {
		//FIXME
		//Task t = (Task) message;
		//return (long) ((10 - t.getCpu()) * t.getDataSize());
		return 0;
	}
*/
	
	public int getNMaxCachedTasks() {
		return NMaxCachedTasks;
	}	
	public void setNMaxCachedTasks(int maxN) {
		NMaxCachedTasks = maxN;
	}	
}

